package com.edata.bigdata.viewmain;

import com.edata.bigdata.basic.Manager;
import com.edata.bigdata.flink.StreamSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/*
 *
 * */
public class FlinkK2C_2 {
    public static void main(String[] args) throws Exception {
        Manager manager = new Manager();
        manager.createFlinkEnvironment();

        /*kafka Source 相关配置*/
        Properties sourceProps = new Properties();
        sourceProps.setProperty("kafka.bootstrap.servers", "172.16.11.97:9092");
        sourceProps.setProperty("subscribe", "test");
        sourceProps.setProperty("group.id", "test");
        sourceProps.setProperty("startingOffsets", "earliest");
        sourceProps.setProperty("kafka.session.timeout.ms", "30000");

        /*管理offset的相关配置，以下是由Kafka自动管理*/
        Properties offsetProps = new Properties();
        offsetProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        offsetProps.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        offsetProps.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        offsetProps.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000");

        StreamSource streamSource = new StreamSource(manager.flinkEnv);
        DataStream<String> sourceStream = streamSource.createDataStream(sourceProps,
                offsetProps);

        sourceStream.print();
        streamSource.start("Kafka2ConsoleFlink");

    }
}
